package drds.global_transaction.api.rpc.rpc_client;

import java.util.concurrent.BlockingQueue;

import drds.global_transaction.api.message_codec.MergeWarpMessage;
import drds.global_transaction.api.message_codec.message_codec.Message;
import drds.global_transaction.api.message_codec.request_and_response.RequestMessage;
import drds.global_transaction.api.message_codec.rpc_message.RpcMessage;
import drds.global_transaction.common.exception.FrameworkErrorCode;
import drds.global_transaction.common.exception.FrameworkException;
import drds.global_transaction.common.util.Nets;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class SendMergedWarpMessageRunnable implements Runnable
{

	private RpcClient rpcClient;

	public SendMergedWarpMessageRunnable(RpcClient rpcClient)
	{
		this.rpcClient = rpcClient;
	}

	@Override
	public void run()
	{
		while (true)
		{
			synchronized (rpcClient.mergedWarpMessageLock)
			{
				try
				{
					rpcClient.mergedWarpMessageLock.wait(RpcClient.max_merge_send_mills);
				} catch (InterruptedException e)
				{
				}
			}
			// mark 'sending message'
			rpcClient.isSending = true;
			for (String address : rpcClient.addressToRpcMessageQueueMap.keySet())
			{
				BlockingQueue<RpcMessage> rpcMessageQueue = rpcClient.addressToRpcMessageQueueMap.get(address);
				if (rpcMessageQueue.isEmpty())
				{
					continue;
				}

				MergeWarpMessage mergedWarpMessage = new MergeWarpMessage();
				while (!rpcMessageQueue.isEmpty())
				{
					RpcMessage rpcMessage = rpcMessageQueue.poll();
					mergedWarpMessage.messageList.add((RequestMessage) rpcMessage.getMessageCodec());
					mergedWarpMessage.rpcMessageIdList.add(rpcMessage.getId());
				}
				if (mergedWarpMessage.rpcMessageIdList.size() > 1)
				{
					printMergeMessageLog(mergedWarpMessage);
				}
				Channel channel = rpcClient.connect(Nets.toInetSocketAddress(address));// 原来是address
				try
				{
					rpcClient.sendRequest(channel, mergedWarpMessage);
				} catch (FrameworkException e)
				{
					if (e.getFrameworkErrorCode() == FrameworkErrorCode.ChannelIsNotWritable && address != null)
					{
						rpcClient.returnChannel(address, channel);
					}
					log.error("", "client merge call failed", e);
				}
			}
			rpcClient.isSending = false;
		}
	}

	private void printMergeMessageLog(MergeWarpMessage mergedWarpMessage)
	{
		if (log.isDebugEnabled())
		{
			log.debug("merge msg size:" + mergedWarpMessage.rpcMessageIdList.size());
			for (Message message : mergedWarpMessage.messageList)
			{
				log.debug(message.toString());
			}
			StringBuilder sb = new StringBuilder();
			for (long l : mergedWarpMessage.rpcMessageIdList)
			{
				sb.append(RpcClient.msg_id_prefix).append(l).append(RpcClient.single_log_postfix);
			}
			sb.append("\n");
			for (long l : rpcClient.rpcMessageIdToRpcMessageFutureMap.keySet())
			{
				sb.append(RpcClient.futures_prefix).append(l).append(RpcClient.single_log_postfix);
			}
			log.debug(sb.toString());
		}
	}
}
